//
// gcssai_job_schedule.cpp
// gcs
// Grand Central Station, controlling process of the Protea Application Server / Business Logic Transactions for SAI Job Scheduling
// Copyright (C) 2003, 2004, 2005 eXegeSys, Inc.
// Copyright (C) 2008 Bruce A. James
//
// This program is part of the Protea Project.
//
// The Protea Project is free software; you can redistribute it and/or modify it 
// under the terms of the GNU General Public License as published by the 
// Free Software Foundation, either version 3 of the License, or (at your 
// option) any later version.
//
// The Protea Project is distributed in the hope that it will be useful, but 
// WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY 
// or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 
// for more details.
// 
// You should have received a copy of the GNU General Public License along 
// with this program.  If not, see <http://www.gnu.org/licenses/>.
// 
// Please send correspondence to:
// 
// theproteaproject@gmail.com
//

#include "../SDK/ToolsInterface.h"
#include "../AppHandler/parms_xml.h"
#include "../AppHandler/message.h"
#include "gcs.h"

//
// Contains the SAI transactions for the Job Schedule
//

//@ SEC 3
//@ CHA Job Schedule Verify
//@ COD JobScheduleVerify
//@ TAB JobScheduleriptor Read
//@ GEN Use JobScheduleVerify to verify field descriptor record

void JobScheduleVerify( XCBuffer* input, verify_type verify_mode, XCTransactionVars* vars, XCGCS* gcs, XCBuffer* orig = NULL ) {
	TRANS_INFUNC( JobScheduleVerify );

	// TODO:  @ SUB ValidateName to check for a valid name
	//ValidateName( FLD_JOB_NAME, input, ERR_JOB, vars );

	//@ SUB DatabaseRetrieve Attempt to retrieve previous record
	bool exists = gcs->DatabaseRetrieve( orig ? orig : input, vars );

	//@ If verifying an ADD transaction,
	if (verify_mode == VERIFY_ADD || verify_mode == VERIFY_COPY) {
		//@ GRS

		//@ If this record already exists,
		//@ ERR C (ALREADY_EXISTS) Job already exists
		if ( exists )
			vars->Errors->Add( ERR_ALREADY_EXISTS, SEV_CRITICAL, FLD_JOB_NAME, 1, ERR_JOB );

		//@ GRE
	} else { // Review, Change or delete
		//@ GRS

		//@ If this record doesn't exist,
		//@ ERR E (DOESNT_EXIST) Job doesn't exist
		if (!exists)
			vars->Errors->Add( ERR_DOESNT_EXIST, SEV_CRITICAL, FLD_JOB_NAME, 1, ERR_JOB );

		//@ GRE
	};

	TRANS_OUTFUNC();
};




//@ SEC 2
//@ CHA AddJob
//@ COD AddJob
//@ NUM 15
TRANSACTION_CODE( XCGCS::AddJob )
{
	XCBuffer* job_schedule = NULL;
	XCBuffer* job_parm = NULL;
	XCBuffer* detail = NULL;

	START_TXN( AddJob );

		//@ Create a job schedule buffer
		job_schedule = App->CreateBuffer( FMT_JOB_SCHEDULE, true );

		//@ Move screen values into job schedule buffer
		job_schedule->MoveCorresponding( txnvars.buffer, txnvars.Errors, true );

		//@ Create a job parameter buffer
		job_parm = App->CreateBuffer( FMT_JOB_SCHEDULE_PARMS, true );

		//@ Create the detail buffer
		detail = App->CreateBuffer( FMT_MAINTAIN_JOB_SCH_PARM, true );

		START_DB_TXN();
			//@ Verify that a record doesn't already exist
			JobScheduleVerify( job_schedule, VERIFY_ADD, &txnvars, this );

			//@ Write the record
			DatabaseAdd( job_schedule, &txnvars );

			//@ Add all database parm records
			if (txnvars.detailcount) {
				job_parm->MoveCorresponding( job_schedule, txnvars.Errors );
				for (int i=0; i < txnvars.detailcount; i++) {
					job_parm->MoveCorresponding( txnvars.detail[i], txnvars.Errors );
					DatabaseAdd( job_parm, &txnvars );

					detail->MoveCorresponding( job_parm, txnvars.Errors, true );
					App->TransResponse( txnvars.result, detail, TRANS_DETAIL_TYPE );
				};
			};
		END_DB_TXN();

		App->TransResponse( txnvars.result, txnvars.buffer, TRANS_HEADER_TYPE );
		txnvars.result->SetSuccess( true );
	CLEANUP_TXN();
		CleanupObjects( 3, &job_schedule, &job_parm, &detail );
	END_TXN();
};




//@ SEC 2
//@ CHA ChangeJob
//@ COD ChangeJob
//@ NUM 15
TRANSACTION_CODE( XCGCS::ChangeJob )
{
	XCBuffer* job_schedule = NULL;
	XCBuffer* job_parm = NULL;
	XCBuffer* detail = NULL;
	XCBuffer* orig_job_schedule = NULL;
	XCDBStatement* stmt = NULL;
	bool Review_Only;

	START_TXN( ChangeJob );

		//@ Create a job schedule buffer
		job_schedule = App->CreateBuffer( FMT_JOB_SCHEDULE, true );

		//@ Move screen values into job schedule buffer
		job_schedule->MoveCorresponding( txnvars.buffer, txnvars.Errors );

		//@ Create a job parameter buffer
		job_parm = App->CreateBuffer( FMT_JOB_SCHEDULE_PARMS, true );

		//@ Create the detail buffer
		detail = App->CreateBuffer( FMT_MAINTAIN_JOB_SCH_PARM, true );

		Review_Only = txnvars.buffer->IsBlank( 2, FLD_JOB_NAME, FLD_APPL_NBR );

		START_DB_TXN();
			if (Review_Only) {
				JobScheduleVerify( job_schedule, VERIFY_REVIEW, &txnvars, this );

				job_parm->MoveCorresponding( job_schedule, NULL );
				txnvars.query->MakeSELECT( job_parm );
				txnvars.query->AddWHERE( 2, job_parm->GetField( FLD_APPL_NBR ), job_parm->GetField( FLD_JOB_NAME ) );
				stmt = txnvars.dbtxn->ExecuteSQL( txnvars.query );
				while (!stmt->Eof()) {
					detail->MoveCorresponding( job_parm, txnvars.Errors, true );
					App->TransResponse( txnvars.result, detail, TRANS_DETAIL_TYPE );
					stmt->MoveNext();
				};
			} else {
				//@ Create a copy of the entered data
				orig_job_schedule = job_schedule->Copy();

				//@ SUB JobScheduleVerify to verify job schedule changes
				JobScheduleVerify( job_schedule, VERIFY_CHANGE, &txnvars, this, orig_job_schedule );

				//@ SUB DatabaseChange to affect changes
				if (job_schedule->MarkChanges( orig_job_schedule ))
					DatabaseChange( job_schedule, orig_job_schedule, &txnvars );

				//@ Delete all database parm records
				job_parm->MoveCorresponding( job_schedule, NULL );
				txnvars.query->MakeDELETE( job_parm );
				txnvars.query->AddWHERE( 2, job_parm->GetField( FLD_APPL_NBR ), job_parm->GetField( FLD_JOB_NAME ) );
				txnvars.dbtxn->ExecuteSQL( txnvars.query, SQL_NO_RESULT );

				//@ Add all database parm records
				if (txnvars.detailcount) {
					for (int i=0; i < txnvars.detailcount; i++) {
						job_parm->MoveCorresponding( txnvars.detail[i], txnvars.Errors );
						DatabaseAdd( job_parm, &txnvars );

						detail->MoveCorresponding( job_parm, txnvars.Errors, true );
						App->TransResponse( txnvars.result, detail, TRANS_DETAIL_TYPE );
					};
				};
			};
		END_DB_TXN();

		txnvars.buffer->MoveCorresponding( job_schedule, txnvars.Errors );
		App->TransResponse( txnvars.result, txnvars.buffer, TRANS_HEADER_TYPE );
		txnvars.result->SetSuccess( true );
	CLEANUP_TXN();
		CleanupObjects( 5, &job_schedule, &orig_job_schedule, &job_parm, &detail, &stmt );
	END_TXN();
};




//@ SEC 2
//@ CHA DeleteJob
//@ COD DeleteJob
//@ NUM 15
TRANSACTION_CODE( XCGCS::DeleteJob )
{
	XCBuffer* job_schedule = NULL;

	START_TXN( DeleteJob );

		//@ Create a job schedule buffer
		job_schedule = App->CreateBuffer( FMT_JOB_SCHEDULE, true );

		//@ Move screen values into job schedule buffer
		job_schedule->MoveField( txnvars.buffer, FLD_JOB_NAME );
		job_schedule->MoveField( txnvars.buffer, FLD_APPL_NBR );

		START_DB_TXN();
			//@ Verify that a record doesn't already exist
			JobScheduleVerify( job_schedule, VERIFY_DELETE, &txnvars, this );

			//@ Update the record
			DatabaseRemove( job_schedule, &txnvars );
		END_DB_TXN();

		txnvars.result->SetSuccess( true );
	CLEANUP_TXN();
		CleanupObjects( 1, &job_schedule );
	END_TXN();
};




//@ SEC 2
//@ CHA ReviewJobSchedule
//@ COD ReviewJobSchedule
//@ NUM 15
TRANSACTION_CODE( XCGCS::ReviewJobSchedule )
{
	XCBuffer* job_schedule = NULL;
	XCBuffer* detail = NULL;
	XCDBStatement* stmt = NULL;

	START_TXN( ReviewJobSchedule );

		//@ Create a job schedule buffer
		job_schedule = App->CreateBuffer( FMT_JOB_SCHEDULE, true );

		START_DB_TXN();
			App->TransResponse( txnvars.result, txnvars.buffer, TRANS_HEADER_TYPE );

			int child = txnvars.buffer->GetChildFormat();
			if (child) {
				//@ Create a detail buffer
				detail = App->CreateBuffer( child, true );

				//@ SUB OrderJobs to get job list
				stmt = OrderJobs( job_schedule, false, &txnvars );
				while (!stmt->Eof()) {
					detail->MoveCorresponding( job_schedule, txnvars.Errors, true );
					App->TransResponse( txnvars.result, detail, TRANS_DETAIL_TYPE );
					stmt->MoveNext();
				};
			};
		END_DB_TXN();

		txnvars.result->SetSuccess( true );
	CLEANUP_TXN();
		CleanupObjects( 3, &job_schedule, &detail, &stmt );
	END_TXN();
};



// Return the jobs in order
XCDBStatement* XCGCS::OrderJobs( XCBuffer* job_schedule, bool active_only, XCTransactionVars* vars ) {
	XCSQLGen* query = vars->query;
	XCOrder order;

	query->MakeSELECT( job_schedule );
	if (active_only) {
		XCField* state = job_schedule->GetField( FLD_JOB_STATE );
		state->SetLiteral( LIT_STATE_ACTIVE );
		query->AddWHERE( 1, state );
	};
	query->AddORDERBY( FLD_JOB_SCHED_DATE, &order );

	return vars->dbtxn->ExecuteSQL( query );
};


struct execute_single_parms {
	XCBuffer* job_schedule;
	char** parameters;
	int parameter_count;
	XCGCS* gcs;
	XCTxn* txn;
};


//
// Writes a job admin message to the system.  All admin messages start with the job name,
// and appl nbr.  The failure message also includes an error message
//
void WriteJobAdminMsg( XCGCS* gcs, XCBuffer* job_schedule, int msg_nbr, const char* message ) {
	char* error_msg = gcs->GetCatalogMsg( msg_nbr, 0 );
	const char* job_name = job_schedule->GetField( FLD_JOB_NAME )->GetStringPtr();
	long job_appl_nbr = job_schedule->GetField( FLD_APPL_NBR )->GetLong( true );

	int length = strlen( error_msg ) + strlen( job_name ) + 10;
	if (message)
		length += strlen( message );

	admin_message* msg = (admin_message*)new char[ sizeof( admin_message ) + length ];
	msg->err_num = ERR_JOB_COMPLETE;
	msg->field_num = 0;
	msg->src_pid = 0;
	sprintf( msg->err_message, error_msg, job_name, job_appl_nbr, message?message:"" );
	free( error_msg );

	gcs->LogAdminMsg( msg );
	delete msg;
};



void XCGCS::ExecuteComplete( XCBuffer* job_schedule, bool success, const char* message, XCBuffer* original, XCTxn* txn ) {
	// The job has completed, so go back to the active state
	if (job_schedule->GetField( FLD_JOB_RECUR_MODE )->CompareLiteral( LIT_RECUR_ON_DEMAND )) {
		job_schedule->GetField( FLD_JOB_STATE )->SetLiteral( LIT_STATE_PAUSED );
	} else {
		// This is a recurring job:

		// Go back to active mode
		job_schedule->GetField( FLD_JOB_STATE )->SetLiteral( LIT_STATE_ACTIVE );

		// Update the sched date to be in the future, based on recur mode
		XCField* recur_mode = job_schedule->GetField( FLD_JOB_RECUR_MODE );
		XCField* sched_date = job_schedule->GetField( FLD_JOB_SCHED_DATE );
		sched_date->SetCurrentDate();
		if (recur_mode->CompareLiteral( LIT_RECUR_DAILY )) {
			sched_date->AddConstDays( 1 );
		} else if (recur_mode->CompareLiteral( LIT_RECUR_MONTHLY )) {
			sched_date->AddConstMonths( 1 );
		} else if (recur_mode->CompareLiteral( LIT_RECUR_YEARLY )) {
			sched_date->AddConstYears( 1 );
		};
	};

	if (success) {
		// Write an admin message
		WriteJobAdminMsg( this, job_schedule, ERR_JOB_COMPLETE, NULL );
	} else {
		// Write an admin message
		WriteJobAdminMsg( this, job_schedule, ERR_JOB_FAILURE, message );
	};

	// Update changes
	try {
		XCTransactionVars txnvars( this, NULL, false );
		if (txn)
			txnvars.dbtxn = txn;
		else
			txnvars.dbtxn = GetDatabase()->CreateTxn();

		DatabaseChange( job_schedule, original, &txnvars );

		if (txn)
			txnvars.dbtxn = NULL;
	} catch( XCToolsError& error ) {
		printf( "UNABLE TO UPDATE JOB INFORMATION\n" );
	};
};



struct txn_job_type {
	XCBuffer* job_schedule;
	int number;
	txn_job_type* next;
} *txn_job_list = NULL;
XCCriticalSection job_txn_cs;


void XCGCS::HandleTransactionResponse( XCTransactionInfo* txn ) {
	int trans_id = txn->GetTransID();

	// Find our transaction, protect with a critical section
	job_txn_cs.Lock();
	txn_job_type* job = txn_job_list;
	if (job->number == trans_id) {
		// Easy, first one in the list...
		txn_job_list = job->next;
	} else {
		// Otherwise, find the parent of the one we want
		while (job->next && job->next->number != trans_id) {
			job = job->next;
		};

		// If we found one, remove it from the list
		if (job->next) {
			txn_job_type* the_one = job->next;
			job->next = job->next->next;
			job = the_one;
		} else
			job = NULL;
	};
	job_txn_cs.UnLock();

	if (job) {
		// Update the completion code
		XCBuffer* job_schedule = job->job_schedule;
		XCBuffer* orig = job_schedule->Copy();

		job_schedule->GetField( FLD_JOB_COMP_CODE )->SetLong( txn->GetSuccess() );

		XCField* result_data = job_schedule->GetField( FLD_JOB_RESULT_DATA );
		char* output_data = result_data->GetOffset();
		*output_data = 0;

		int len = result_data->GetLength();

		// Create the output data
        XCMessageList* temp = txn->GetFirstMessage();

        while (temp) {
            XCToolsMessage* msg = temp->msg;
            base_message* header = msg->GetHeaderPtr();
            trans_message* trans_info = (trans_message*)msg->GetDataPtr();

			// Output all error messages
			if (trans_info->data_type == TRANS_MESSAGE_TYPE) {
				trans_error_message* error = (trans_error_message*)trans_info->data;

				// Output the error message
				int written = snprintf( output_data, len, "ERROR: %s\n", error->text );
				len -= written;
				output_data += written;
            }

			temp = temp->next;
		};

		// Mark the transaction as complete
		bool success = txn->GetSuccess();
		ExecuteComplete( job_schedule, success, success?NULL:result_data->GetOffset(), orig );
		orig->Release();

		// Cleanup
		job_schedule->Release();
		delete job;
	};
};



THREAD_PROCEDURE XCGCS::ExecuteSingleJob( void* parms ) {
	execute_single_parms* exec_parms = (execute_single_parms*)parms;
	XCBuffer* job_schedule = exec_parms->job_schedule;
	XCBuffer* orig = job_schedule->Copy();
	XCField* result_data = job_schedule->GetField( FLD_JOB_RESULT_DATA );
	bool success = true;
	bool send_message = true;

	XCTransactionVars txnvars( exec_parms->gcs, NULL, false );
	try {
		// Create/reuse transaction
		if (exec_parms->txn)
			txnvars.dbtxn = exec_parms->txn;
		else
			txnvars.dbtxn = exec_parms->gcs->GetDatabase()->CreateTxn();

		try {
			// Record date/time that job ran
			job_schedule->GetField( FLD_JOB_RUN_DATE )->SetCurrentDate();
			job_schedule->GetField( FLD_JOB_RUN_TIME )->SetCurrentDate();
			job_schedule->GetField( FLD_JOB_STATE )->SetLiteral( LIT_STATE_RUNNING );

			// Update changes
			exec_parms->gcs->DatabaseChange( job_schedule, orig, &txnvars );
		} catch( ... ) {
			if (!exec_parms->txn)
				txnvars.dbtxn->EndTxn( false );
			txnvars.dbtxn = NULL;
			throw;
		};
		if (!exec_parms->txn)
			txnvars.dbtxn->EndTxn( true );
		txnvars.dbtxn = NULL;

		// Write admin message
		WriteJobAdminMsg( exec_parms->gcs, job_schedule, ERR_JOB_STARTED, NULL );

		// Restart a new transaction
		orig->MoveCorresponding( job_schedule, NULL );

		// Now, run the job...
		XCField* job_type = job_schedule->GetField( FLD_JOB_TYPE );
		if (job_type->CompareLiteral( LIT_JOB_TYPE_PREP )) {
			// Execute prep...
			char parm[ 255 ];
			char** parms = new char*[2];
			parms[0] = parm;
			parms[1] = NULL;
			sprintf( parm, "%d", job_schedule->GetField( FLD_APPL_NBR )->GetLong( true ) );
			int ret_code = XCExecuteAndWait( "prep", 1, parms );
			job_schedule->GetField( FLD_JOB_COMP_CODE )->SetLong( ret_code );
		} else if (job_type->CompareLiteral( LIT_JOB_TYPE_TXN )) {
			XCTransactionInfo* info = new XCTransactionInfo( exec_parms->gcs, NULL );
			int app_nbr = job_schedule->GetField( FLD_APPL_NBR )->GetLong( true );
			XCApplication* app = exec_parms->gcs->hostlist.LoadBalance( app_nbr );
			if (app) {
				info->SetTransID( rand() ); // Set transaction id
				info->SetTransNumber( atoi( *exec_parms->parameters ) );
				info->SetUser( "system" );
				info->SetPass( "system", info->GetTransID() );
				info->SetLang( 0 );
				info->SetSuperID( 0 );
				info->SetAction( 0 );
				info->SetJob();

				// Add the job to the job list
				txn_job_type* job = new txn_job_type;
				job->number = info->GetTransID();
				job->job_schedule = job_schedule;
				job_schedule = NULL;

				// Protect with a critical section
				job_txn_cs.Lock();
				job->next = txn_job_list;
				txn_job_list = job;
				job_txn_cs.UnLock();

				exec_parms->gcs->SendTransaction( info, app->owner->msg_port, NULL, app_nbr, src_gcs );
				send_message = false;
			} else {
				success = false;
				result_data->SetString( "Application not running" );
			};
		} else if (job_type->CompareLiteral( LIT_JOB_TYPE_EXTERN )) {
			// Execute the external program
			char* program = exec_parms->parameters[0];
			int ret_code = XCExecuteAndWait( program, exec_parms->parameter_count-1, exec_parms->parameters+1 );
			job_schedule->GetField( FLD_JOB_COMP_CODE )->SetLong( ret_code );
		} else {
			success = false;
			result_data->SetString( "Execution type not supported!\n" );
		};
	} catch( XCToolsError& error ) {
		printf( "CAUGHT TOOLS ERROR\n" );

		// Store the error message
		result_data->SetString( error.GetDescription() );
		job_schedule->GetField( FLD_JOB_COMP_CODE )->SetLong( error.GetErrorNumber() );

		// Signal that we were not successful
		success = false;
	} catch( ... ) {
		// Store the error message
		result_data->SetString( "Unknown error" );
		job_schedule->GetField( FLD_JOB_COMP_CODE )->SetLong( -1 );

		// Signal that we were not successful
		success = false;
	};

	// Send back job execution status...
	if (send_message) {
		exec_parms->gcs->ExecuteComplete( job_schedule, success, success?NULL:result_data->GetOffset(), orig, exec_parms->txn );
	};

	orig->Release();
	if (job_schedule)
		job_schedule->Release();

	if (exec_parms->parameters) {
		for (int i=0; i<exec_parms->parameter_count; i++) {
			free( exec_parms->parameters[i] );
		};

		delete [] exec_parms->parameters;
	};

	delete exec_parms;

	return 0;
};



void XCGCS::ExecuteJobs() {
	// Check license
	const long SECURITY_CHECKER = 100;
	static int counter = SECURITY_CHECKER;

	counter--;
	if (counter < 0) {
		try {
			counter = SECURITY_CHECKER;
			DO_TRAFFIC_SECURITY_CHECK;
		} catch( ... ) {
			CancelMessageLoop();
			throw;
		};
	};

	XCDBStatement* stmt = NULL;
	XCDBStatement* stmt2 = NULL;
	XCBuffer* job_schedule = NULL;
	XCBuffer* job_parms = NULL;
	XCBuffer* count_buffer = NULL;
	XCField* current_date = NULL;
	XCField* current_time = NULL;

	XCOrder order;
	XCTransactionVars txnvars( this, NULL, false );
	try {
		bool process_more_records = true;

		txnvars.dbtxn = GetDatabase()->CreateTxn();
		XCSQLGen* query = txnvars.query;

		// Return the list of jobs to be executed
		job_schedule = CreateBuffer( FMT_JOB_SCHEDULE );
		stmt = OrderJobs( job_schedule, true, &txnvars );
		
		XCField* sched_date = job_schedule->GetField( FLD_JOB_SCHED_DATE );
		XCField* sched_time = job_schedule->GetField( FLD_JOB_SCHED_TIME );
		current_date = sched_date->Clone();
		current_time = sched_time->Clone();
		current_date->SetCurrentDate();
		current_time->SetCurrentDate();

		job_parms = CreateBuffer( FMT_JOB_SCHEDULE_PARMS );

		double current_jul = current_date->ToJulian() + current_time->ToJulian();
		double sched_jul;
		while (!stmt->Eof() && process_more_records) {
			sched_jul = sched_date->ToJulian() + sched_time->ToJulian();

			//@ If the sched time is before current time:
			if (current_jul > sched_jul) {
				//@ GRS

				//@ Prepare the thread parameters
				execute_single_parms* parms = new execute_single_parms;
				parms->gcs = this;
				parms->job_schedule = job_schedule->Copy();

				//@ Count the job parameters
				job_parms->MoveCorresponding( job_schedule, NULL );
				count_buffer = query->MakeCOUNT( job_parms );
				query->AddWHERE( 2, job_parms->GetField( FLD_APPL_NBR ), job_parms->GetField( FLD_JOB_NAME ) );
				txnvars.dbtxn->ExecuteSQL( query, SQL_NO_RESULT );

				parms->parameter_count = count_buffer->GetFieldByIndex(0)->GetLong( true );
				if (parms->parameter_count) {
					parms->parameters = new char*[ parms->parameter_count+1 ];

					query->MakeSELECT( job_parms );
					query->AddWHERE( 2, job_parms->GetField( FLD_APPL_NBR ), job_parms->GetField( FLD_JOB_NAME ) );
					query->AddORDERBY( FLD_JOB_PARM_NUMBER, &order );
					stmt2 = txnvars.dbtxn->ExecuteSQL( query );

					//@ Read the job parameters
					for (int i=0; i<parms->parameter_count; i++) {
						XCField* value = job_parms->GetField( FLD_JOB_PARM_VALUE );
						value->CalcStringSize();
						parms->parameters[i] = strdup( value->GetStringPtr() );
						stmt2->MoveNext();
					};
					parms->parameters[ parms->parameter_count ] = NULL;

					CleanupObjects( 1, &stmt2 );
				} else
					parms->parameters = NULL;

				//@ SUB ExecuteSingleJob Create a thread to run the execution
//				parms->txn = txnvars.dbtxn;
//				ExecuteSingleJob( parms );

				parms->txn = NULL;
				XCCreateThread( ExecuteSingleJob, parms );

				//@ Move on to next record
				stmt->MoveNext();
			} else
				process_more_records = false;
		};
	} catch( ... ) {
		CleanupObjects( 7, &stmt, &stmt2, &job_schedule, &job_parms, &current_date, &current_time, &count_buffer );
		txnvars.dbtxn->EndTxn( false );
		throw;
	};
	CleanupObjects( 7, &stmt, &stmt2, &job_schedule, &job_parms, &current_date, &current_time, &count_buffer );
	txnvars.dbtxn->EndTxn( true );
};

